package com.dmp.userLabel.bolt;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import com.dmp.storm.base.BaseMapKey;
import com.dmp.userLabel.factory.ConcreteFactory;
import com.dmp.userLabel.factory.Factory;
import com.dmp.userLabel.factory.ILabelRedisDao;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;

/**
 * Created by wangqiang on 2016/10/31.
 */
public class KafkaBolt extends BaseRichBolt {
    private static final long serialVersionUID = 1L;
    protected OutputCollector collector;
    Logger logger = LoggerFactory.getLogger(KafkaBolt.class);

    private String splitChars;
    private String redisPath;
    private ILabelRedisDao iLabelRedisDao;

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        //创建一个具体工厂
        Factory factory = new ConcreteFactory();
        iLabelRedisDao = factory.createLabelRedisDao(this.redisPath);
        this.collector = collector;
    }

    @Override
    public void execute(Tuple tuple) {
        try {
            long now = new Date().getTime();
            Map<String, String> resultMap = new HashMap<>();
            String data = tuple.getStringByField("msg");
            if (StringUtils.isNotBlank(data)) {
                String[] values = data.split(splitChars);
                if (values.length > 0) {
                    resultMap.put(BaseMapKey.MAPKEY_RESULTMAP_BOLTMSG, data);
                    if (resultMap != null) {
                        collector.emit(tuple, new Values(resultMap, ""));
                    }
                    collector.ack(tuple);
                }

            }


        } catch (Exception e) {
            e.printStackTrace();
            System.err.println(this.getClass().getSimpleName() + "  KafkaBolt:execute=" + e.getMessage());
            collector.fail(tuple);
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields(BaseMapKey.MAPKEY_TUPLEBOLT_RESULTMAP, "keyword"));
    }


    public String getSplitChars() {
        return splitChars;
    }

    public void setSplitChars(String splitChars) {
        this.splitChars = splitChars;
    }

    public String getRedisPath() {
        return redisPath;
    }

    public void setRedisPath(String redisPath) {
        this.redisPath = redisPath;
    }

}
